package com.hujiang.basic.search.fullindex;

import com.alibaba.fastjson.JSON;
import com.hujiang.basic.search.common.exception.FullIndexException;
import com.hujiang.basic.search.common.service.NotifyService;
import com.hujiang.basic.search.common.utils.Assert;
import com.hujiang.basic.search.handler.impl.model.FullIndexData;
import com.hujiang.basic.search.handler.impl.model.FullIndexResponse;
import com.hujiang.basic.search.handler.impl.model.FullIndexTask;
import com.hujiang.basic.search.handler.impl.model.TaskState;
import com.hujiang.basic.search.util.IndexEsDao;
import com.hujiang.basic.search.util.IpUtils;
import com.hujiang.basic.search.zookeeper.zkLock.DistributedLock;
import com.hujiang.basic.search.zookeeper.zkNotify.DistributedNotifyClient;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.message.BasicHeader;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class TaskProcessor {

    @Autowired
    private IndexEsDao esDao;

    @Autowired
    private TaskDao taskDao;

    @Autowired
    private NotifyService notifyService;

    @Autowired
    private DistributedNotifyClient notifyClient;

    @Autowired
    private DistributedLock lock;

    private static final int MAX_RETRY_CNT = 3;
    private static final int HTTP_CONNECT_TIMEOUT = 20 * 1000;
    private static final double MIN_INDEX_PERCENTAGE = 0.7;
    private static final int FULLINDEX_LOCK_WAIT_TIME = 5;

    @Scheduled(fixedDelay = 1_000)
    public void start() {
        FullIndexTask task = taskDao.findExpiredTask().stream().findFirst().orElse(null);    // 查找处理超时的任务
        task = task == null ? taskDao.findSpareTask().stream().findFirst().orElse(null) : task; // 查找空闲的任务
        if (task == null) {// 如果没有任务，则退出本次定时任务
            return;
        }

        if (lock.fullIndexLocks.containsKey(task.getAlias())) {
            InterProcessMutex fullIndexlock = this.lock.fullIndexLocks.get(task.getAlias());
            try {
                if (this.lock.fullIndexLocks.get(task.getAlias()).acquire(FULLINDEX_LOCK_WAIT_TIME, TimeUnit.SECONDS)) {
                    fullIndexing(task);
                }
            } catch (Exception ex) {
                log.error(ex.getMessage(), ex);
            } finally {
                try {
                    fullIndexlock.release();
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        } else {//适配暂时未配置queue的索引
            fullIndexing(task);
        }
    }

    private void fullIndexing(FullIndexTask task) {
        if (task.getState() == TaskState.Start) {
            log.info("接管全量任务: {}", task.toString());
        } else {
            task.setState(TaskState.Start);
            task.setRunningNode(IpUtils.getNodeAddr());
            taskDao.acceptTask(task);
        }
        log.info("领取任务成功，执行全量导入任务{} ", task);
        boolean hasZkNotifyPath = notifyClient.distributedNotifyMap.containsKey(task.getAlias());
        try {
            if (hasZkNotifyPath) {
                notifyClient.distributedNotifyMap.get(task.getAlias()).setValue("true");
            }
            while (task.getState() == TaskState.Start) {
                FullIndexData data = fetchDataWithRetry(task, MAX_RETRY_CNT);
                saveToEs(task, data);
                refreshTask(task, data);
                taskDao.updateTask(task);
            }
        } catch (Exception ex) { //maybe allocate error because no more disk when close index
            log.error("TaskProcessor fetch data and save to es error ",ex);
            taskDao.updateTask(task);
            esDao.closeIndex(task.getNewIndex());
        } finally {
            if (hasZkNotifyPath) {
                try {
                    notifyClient.distributedNotifyMap.get(task.getAlias()).setValue("false");
                } catch (Exception e) {
                    log.error("distributedNotify  set index stop full index error:{}",e);
                    log.error(e.getMessage());
                }
            }
        }
        notifyService.inform(task.getOwner(), task.toString());
    }


    private FullIndexData fetchDataWithRetry(FullIndexTask task, int maxRetry) {
        while (task.getRetryCount() < maxRetry) {
            try {
                return getFullIndexData(task);
            } catch (Exception e) {
                task.setRetryCount(task.getRetryCount() + 1);
                task.setMessage(StringUtils.substring(e.getMessage(), 0, 1000));
                log.error("encounter exception during full index data", e);
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e1) {
                    log.error(e1.getMessage(), e1);
                }
            }
        }
        if (task.getRetryCount() >= maxRetry) {
            task.setState(TaskState.Failed);
            esDao.closeIndex(task.getNewIndex());
        }
        return null;
    }

    private void refreshTask(FullIndexTask task, FullIndexData data) {
        if (task.getState() == TaskState.Failed) {
            return;
        }
        if (isTaskCompleted(data)) {
            task.setState(TaskState.Finished);
            return;
        }
        task.setFrom(data.getFrom());
        task.setCustomData(data.getCustomData());
        task.setSize(calcBulkSize(task));
        task.setRetryCount(0);
    }

    private boolean isTaskCompleted(FullIndexData data) {
        return data == null || data.getIndexData() == null || CollectionUtils.isEmpty(data.getIndexData().getSources());
    }

    @Value("${fullIndex.defaultWindowSize:5000}")
    private int bulkSize;

    private int calcBulkSize(FullIndexTask task) {
        return 0 < task.getSize() ? task.getSize() : bulkSize;
    }

    private FullIndexData getFullIndexData(FullIndexTask task) throws IOException {
        String url = buildUrl(task);
        long start = System.currentTimeMillis();
        List<Header> headers = Arrays.asList(new BasicHeader("Accept-Encoding", "GZip,Deflate"));
        RequestConfig requestConfig = RequestConfig.custom().setConnectionRequestTimeout(HTTP_CONNECT_TIMEOUT).build();
        log.info("Start to get index({}) data({}) from url({}).", task.getNewIndex(), task.getSize(), url);
        try (CloseableHttpClient httpClient = HttpClientBuilder.create()
                .setDefaultHeaders(headers)
                .setDefaultRequestConfig(requestConfig).build();
             CloseableHttpResponse httpResp = httpClient.execute(new HttpGet(url))) {
            Header header = httpResp.getFirstHeader("charset");
            String charset = header != null ? header.getValue() : "UTF-8";
            HttpEntity httpEntity = httpResp.getEntity();
            String json = EntityUtils.toString(httpEntity, charset);
            FullIndexResponse rsResp = JSON.parseObject(json, FullIndexResponse.class);
            if (rsResp.getStatus() != 0) {
                throw new FullIndexException("Response status: " + rsResp.getStatus() + ", message: " + rsResp.getMessage());
            }
            log.info("End({}) to get index({}) data({}) from url({}).",
                    (new Date().getTime() - start), task.getNewIndex(), task.getSize(), url);
            return rsResp.getData();
        } catch (Exception e) {
            log.error("End({}) to get index({}) data({}) from url({}),error:{}.",
                    (new Date().getTime() - start), task.getNewIndex(), task.getSize(), url, e);
            throw e;
        }
    }

    private String buildUrl(FullIndexTask task) {
        String url = task.getUrl() + "?from=" + task.getFrom() + "&size=" + task.getSize() + "&id=" + task.getTaskId();
        if (!StringUtils.isBlank(task.getCustomData())) {
            url = url + "&customData=" + task.getCustomData();
        }
        return url;
    }

    private void saveToEs(FullIndexTask task, FullIndexData data) {
        if (task.getState() == TaskState.Failed) {
            return;
        }
        try {
            String alias = task.getAlias();
            String oldIndex = task.getIndex();
            String newIndex = task.getNewIndex();

            log.info("saveToEs, alias: {}, oldIndex: {}, newIndex: {}", alias, oldIndex, newIndex);

            if (isTaskCompleted(data)) {
                String index = esDao.getIndexName(alias);
                long oldTotalCount = esDao.getCount(index);
                esDao.flushAndRefresh(newIndex);
                long newTotalCount = esDao.getCount(newIndex);
                if (newTotalCount < oldTotalCount * MIN_INDEX_PERCENTAGE) {
                    log.info("新索引数据少于原索引70%，new index size :{}，old index size：{}", newTotalCount, oldTotalCount);
                    notifyService.inform(task.getOwner(), "新索引数据少于原索引70%,不切换索引");
                    return;
                }
                task.setState(TaskState.Finished);
                esDao.changeIndexAlias(oldIndex, newIndex, alias);
            } else if (TaskState.Start == task.getState()) {
                String actual = data.getIndexData().getIndex();
                Assert.equalForObject(alias, actual, "索引不匹配：expected:" + alias + ", actual:" + actual);
                esDao.addBulkIndexData(newIndex, data.getIndexData());
            }

        } catch (Exception e) {
            task.setState(TaskState.Failed);
            task.setMessage("保存ES出错: " + e.getMessage() + "+++" + task.getMessage());
            log.error("保存数据到ES失败，任务退出：" + task.errorClues(), e);
            esDao.closeIndex(task.getNewIndex());
        }
    }
}
